package SparkRDD.RDD的CheckPoint

import org.apache.commons.lang.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

/**
 *  CheckPoint 的主要作用是斩断 RDD 的依赖链，并将数据存储在可靠的存储引擎中
 *
 *  CheckPoint 的方式：
 *        可靠的 --- 将数据存储在可靠的存储引擎中，例如HDFS
 *        本地的 --- 将数据存储在本地
 *
 */

class checkPointTest {

  @Test
  def test: Unit ={

    val conf = new SparkConf().setAppName("ip统计").setMaster("local[6]")
    val sc   = new SparkContext(conf)
    // 设置保存 checkPoint 的目录
    sc.setCheckpointDir("checkpoint")

    val resource = sc.textFile("src/main/scala/RDD的缓存/ip.txt")
    val ipRDD = resource.map( item => ( item.split(",")(0) , 1) )
    val cleanRDD = ipRDD.filter( item => StringUtils.isNotEmpty(item._1) )
    var aggRDD = cleanRDD.reduceByKey( (curr,agg) => curr + agg )

    // checkPoint
    // aggRDD = aggRDD.cache()
    // 不准确的说 checkPoint 是一个Action操作，
    // 也就是说调用了 checkPoint 方法后会重新计算一遍RDD的值，然后把结果存在HDFS或本地目录中
    // 所以因该在 checkPoint 之前进行一次 cache()
    aggRDD = aggRDD.cache()
    aggRDD.checkpoint()

    val maxRDD = aggRDD.sortBy( item => item._2,ascending = true ).first()   // Action操作1
    val minRDD = aggRDD.sortBy( item => item._2,ascending = false ).first()  // Action操作2
    println("max:"+maxRDD,"min:"+minRDD)

  }

}
